package com.xpec.c4.service.group.transportlayer;

import java.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;

import javax.management.MBeanServer;
import javax.management.MBeanServerFactory;
import javax.management.ObjectInstance;
import javax.management.ObjectName;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jgroups.Address;
import org.jgroups.JChannel;
import org.jgroups.MembershipListener;
import org.jgroups.Message;
import org.jgroups.SuspectedException;
import org.jgroups.TimeoutException;
import org.jgroups.View;
import org.jgroups.blocks.MessageDispatcher;
import org.jgroups.blocks.RequestHandler;
import org.jgroups.blocks.RequestOptions;
import org.jgroups.blocks.ResponseMode;
import org.jgroups.conf.ProtocolStackConfigurator;
import org.jgroups.jmx.JmxConfigurator;

import com.ms.service.module.commonsettings.CommonSettings;
import com.xpec.c4.codec.CodecUtility;
import com.xpec.c4.game.time.JobExecutor;
import com.xpec.c4.game.time.TimeConstants;
import com.xpec.c4.game.utility.GUID;
import com.xpec.c4.game.utility.GameUtility;
import com.xpec.c4.h5net.H5XSENetSession;
import com.xpec.c4.message.ContextStatus;
import com.xpec.c4.message.MessageTypes;
import com.xpec.c4.net.XSENetSession;
import com.xpec.c4.service.common.MessageHandle;
import com.xpec.c4.service.common.ServiceInfo;
import com.xpec.c4.service.contexts.Contexts;
import com.xpec.c4.service.exceptions.GroupTransportException;
import com.xpec.c4.service.group.JgroupUtitly;
import com.xpec.c4.service.h5common.H5ServiceInfo;
import com.xpec.c4.service.h5server.H5ServerOperations;
import com.xpec.c4.service.message.MessageAction;
import com.xpec.c4.service.module.BasicServiceModule;
import com.xpec.c4.service.module.ModuleName;
import com.xpec.c4.service.module.ModuleServer;
import com.xpec.c4.service.module.ServerProperties;
import com.xpec.c4.service.module.SyncSignalManager;
import com.xpec.c4.service.module.common.PlayerCommonInfoManager;
import com.xpec.c4.service.module.gateway.GatewayModule;
import com.xpec.c4.service.module.message.ModuleStatus;
import com.xpec.c4.service.server.CloseType;
import com.xpec.c4.service.server.ServerOperations;

/**
 * 组通讯管理
 * 
 * @author lipeilin
 * @2011-3-10 下午02:36:39
 * 
 */
public class TransportManager {

	private static final Log log = LogFactory.getLog(TransportManager.class);


	/**
	 * 通道
	 */
	private JChannel channel;

	/**
	 * 消息分发
	 */
	private MessageDispatcher dispatcher;

	/**
	 * 成员信息列表
	 */
	private ConcurrentHashMap<Address, GroupMember> members = new ConcurrentHashMap<Address, GroupMember>();

	/**
	 * 模块地址列表
	 */
	private ConcurrentHashMap<String, Address> moduleAddress = new ConcurrentHashMap<String, Address>();

	/**
	 * 模块Class對應模块名稱列表
	 */
	private ConcurrentHashMap<String, Set<String>> moduleClassMap = new ConcurrentHashMap<String, Set<String>>();
	
	/**
	 * 同步訊號管理
	 */
	private SyncSignalManager syncSignalManager = new SyncSignalManager();
	
	/**
	 * 模块ID与名称映射
	 */
	// private HashMap<Integer, String> moduleIdNames = new HashMap<Integer,
	// String>();
	/**
	 * 模块服务器
	 */
	private ModuleServer moduleServer;

	/**
	 * session过滤器
	 */
	@SuppressWarnings("unused")
	private ISessionFilter filter;

	private boolean running = false;

	private ObjectName jgroupsChannelMBeanName[] = new ObjectName[2];

	/**
	 * 是否已經呼叫過 {@link BasicServiceModule#onServerConnectReady()} 通知事件
	 */
	private boolean serverReadyNotified = false;

	/**
	 * 指定特定廣播消息只送給部份 module 的 map<br>
	 * key: 廣播消息編號<br>
	 * value 指定要接收的 module id 列表
	 */
	private HashMap<Integer, HashSet<String>> broadcastMsgListenModuleMap = new HashMap<Integer, HashSet<String>>();

	/**
	 * 
	 */
	public TransportManager(ModuleServer server) {

		this.moduleServer = server;
/*		
		try {
			this.filter = (ISessionFilter) Class.forName(
					GroupProperties.sessionFilterClass).newInstance();
		} catch (InstantiationException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
*/		
		registerBroadcastMsgListenModuleMap();
	}
	
	/**
	 * 註冊需要收到特定廣播消息的 module
	 */
	private void registerBroadcastMsgListenModuleMap(){
		// 角色登入通知
		int messageType = MessageTypes.S_ACCOUNT_CHARACTER_LOGIN_NOTIFY;
		HashSet<String> listenModuleSet = new HashSet<String>();
		
		// BM test 暫時讓紅包 module 收登入廣播，不然紅包不會運作
		listenModuleSet.add(ModuleName.HONBO);
		
		broadcastMsgListenModuleMap.put(messageType, listenModuleSet);
		
		// 角色斷線通知
		messageType = MessageTypes.S_SESSION_CLOSE_NOTIFY;
		listenModuleSet = new HashSet<String>();
		listenModuleSet.add(ModuleName.ACCOUNT);
		listenModuleSet.add(ModuleName.LOGIN);
		listenModuleSet.add(ModuleName.GAMELOBBY);
		broadcastMsgListenModuleMap.put(messageType, listenModuleSet);
	}

	/**
	 * 获得全部组成员信息
	 * 
	 * @return
	 * 
	 * @author lipeilin
	 * @2011-3-10 下午03:42:41
	 */
	public List<GroupMember> getMembers() {
		return new ArrayList<GroupMember>(members.values());
	}

	/**
	 * 获得全部节点地址
	 * 
	 * @return
	 * 
	 * @author lipeilin
	 * @2011-3-10 下午03:42:41
	 */
	public List<Address> getAllAddress() {
		return new ArrayList<Address>(members.keySet());
	}

	/**
	 * 检查本节点是否协调节点
	 * 
	 * @return
	 * 
	 * @author lipeilin
	 * @2011-3-18 下午07:08:09
	 */
	public boolean isCoordinator() {
		List<Address> mbrs = channel.getView() != null ? channel.getView()
				.getMembers() : null;
		if (mbrs == null) {
			return false;
		}
		if (!mbrs.isEmpty()) {
			Address address = mbrs.get(0);
			if (address.equals(channel.getAddress())) {
				return true;
			}
			return false;
		}
		return false;
	}

	/**
	 * 获得本地信息
	 * 
	 * @author lipeilin
	 * @2011-3-10 下午03:43:18
	 */
	public GroupMember getLocal() {
		Address address = channel.getAddress();
		GroupMember member = members.get(address);
		if (member == null) {
			member = initLocalMemberInfo();
		}
		return member;
	}

	/**
	 * 获得本地信息
	 * 
	 */
	public List<String> getLocalModuleList() {
		return getLocal().getServiceModule();
	}

	public void startGroupTransport() throws Exception {

		if (running) {
			log.error("group transport already running!");
			return;
		}

		ProtocolStackConfigurator configurator = null;
		String protocol = ServerProperties.getString(
				"group_transport_protocol", "TCP");

		// TCP
		if (protocol.toUpperCase().equals("TCP")) {
			configurator = JgroupUtitly.getJgroupsTCPConfigurator(
					ServerProperties.groupIP,
					ServerProperties.groupTransportPort);
		}
		else if (protocol.toUpperCase().equals("TCPPING")) {
			configurator = JgroupUtitly.getJgroupsTcpPingTranConfigurator();
		}
		// TCP_GOSSIP
		else if (protocol.toUpperCase().equals("TCPGOSSIP")) {
				configurator = JgroupUtitly.getJgroupsTcpGossipConfigurator();
		}
		else
		{
			configurator = JgroupUtitly.getJgroupsUDP2Configurator(
					ServerProperties.groupIP,
					ServerProperties.groupTransportPort);
		}

		// mcast_port
		channel = new JChannel(configurator);
		channel.setName(this.moduleServer.getName() + "_transport");
		// GroupMessageListener messageListener = new GroupMessageListener();
		GroupMembershipListener memberListener = new GroupMembershipListener();
		RequestHandler requestHandler = new GroupMessageHandler(moduleServer);
		dispatcher = new MessageDispatcher(channel, null, memberListener,requestHandler);
		log.info("ProtocolStackString:" + configurator.getProtocolStackString());

		log.error("JGroups start connect to cluster group...");
		
		/**↓very slow ↓**/
		channel.connect("MazeMythServerGroup");
	
		
		while (!channel.isConnected()) {
			Thread.sleep(1000);
			log.error("waiting for Transport connect ... ");
		}

		while (!channel.isOpen()) {
			Thread.sleep(1000);
			log.error("waiting for Transport open ... ");
		}
		
		log.error("Transport Channel: " + channel.getAddress() + " "
				+ ServerProperties.groupTransportPort);		

		initLocalMemberInfo();

		running = true;

		//
		List<BasicServiceModule> list = new ArrayList<BasicServiceModule>(ModuleServer.getInstance().getModules());

		for (BasicServiceModule module : list) {
			if (module.getModuleName().startsWith(ModuleName.GATEWAY)
					&& module instanceof GatewayModule) {
				GatewayModule gatewayModule = (GatewayModule) module;
				gatewayModule.notifyAccount();
			}
		}

		setupJMXBeanServer();		
	}

	private void setupJMXBeanServer() throws Exception {
		//Rowan: add Mbean server here
		//[1]register
		ArrayList<?> servers = MBeanServerFactory.findMBeanServer(null);
		MBeanServer server=(MBeanServer)servers.get(0);
		JmxConfigurator.registerChannel(channel, server, "JGroups:channel=" + channel.getName());
		
		Class<?> transportProtoClass = Class.forName("org.jgroups.protocols.TP");
		Class<?> channelClass = Class.forName("org.jgroups.JChannel");
		//[2]get Channel and Transport Protocol 
		Set<ObjectInstance> objectInstances = server.queryMBeans(null,null);
		for(ObjectInstance objInst : objectInstances)
		{
			if(channelClass.isAssignableFrom(Class.forName(objInst.getClassName())))
			{
				jgroupsChannelMBeanName[0] = objInst.getObjectName();
			}
			
			if(transportProtoClass.isAssignableFrom(Class.forName(objInst.getClassName())))
			{
				jgroupsChannelMBeanName[1] = objInst.getObjectName();	
			}
		}
		
		server = ManagementFactory.getPlatformMBeanServer();
		GroupMessageStat bean = new GroupMessageStat();
		 
		try 
		{
			ObjectName name = new ObjectName("TransportManager:name="+this.toString());
			server.registerMBean(bean, name);
		} 
		catch (Exception e)
		{
			log.error(e, e);
			throw new RuntimeException(e);
		}		
	}
	
	private String channelAttributeNames[] = 
		{
			"received_bytes","received_messages","sent_bytes","sent_messages",//for channel
		};
	
	private String protocolAttributeNames[] = 
		{
			"num_bytes_received","num_bytes_sent","num_msgs_received","num_msgs_sent","num_incoming_msgs_received",//for transport protocol
		};
	
	/**
	 * @return
	 */
	public synchronized Map<String,Integer> getStat()
	{
		Map<String,Integer> map = new TreeMap<String,Integer>();
		//assume the 0th server
		MBeanServer server=(MBeanServer)(MBeanServerFactory.findMBeanServer(null).get(0));
		
		try
		{
			for(int i=0;i<channelAttributeNames.length;++i)
			{
				//update stat and get delta value
				long val = (Long)server.getAttribute(jgroupsChannelMBeanName[0], channelAttributeNames[i]);
				map.put(channelAttributeNames[i], (int)val);
			}
			
			String protoName = (String)server.getAttribute(jgroupsChannelMBeanName[1], "name");
			for(int i=0;i<protocolAttributeNames.length;++i)
			{
				//update stat and get delta value
				long val = (Long)server.getAttribute(jgroupsChannelMBeanName[1], protocolAttributeNames[i]);
				map.put(protoName + "->" + protocolAttributeNames[i], (int)val);
			}			
		} 
		catch (Exception e) 
		{
			log.error(e, e);
		}
		
		return map;
	}
	/**
	 * 设置本地的成员信息
	 * 
	 * 
	 * @author lipeilin
	 * @2011-3-10 下午06:07:25
	 */
	private GroupMember initLocalMemberInfo() {
		// 添加本地信息
		Address address = channel.getAddress();
		List<BasicServiceModule> Modules = new ArrayList<BasicServiceModule>(moduleServer.getModules());
		GroupMember gm = new GroupMember();
		gm.setAddress(address.toString());
		gm.setServerName(moduleServer.getName());
		gm.setLoginTime(System.currentTimeMillis());
		for (BasicServiceModule module : Modules) {
			gm.getServiceModule().add(module.getModuleName());
		}

		addMemberInfo(address, gm);
		return gm;
	}

	/**
	 * 通知目前這個 java vm 底下的 module jgdoup 的 member 都已經加進來 
	 */
	public synchronized void notifyServerReady() {
		// 檢查實際數量與設定檔中的數量
		if (moduleAddress.size() == ServerProperties.getTotalModuleCount()) {
			// 本地 vm 的所有 module 初始話完成
			if (ModuleServer.getInstance().isReady()) {
				// 只做一次
				if (!serverReadyNotified) {
					if (log.isInfoEnabled()) {
						log.info("group member connect ready.");
					}

					// 建立判定用的檔案
					ModuleServer.getInstance().createServerReadySignalFile();

					// 執行  module 事件
					List<BasicServiceModule> modules = new ArrayList<BasicServiceModule>(moduleServer.getModules());
					for (BasicServiceModule module : modules) {
						module.onServerConnectReady();
					}

					// 送給此vm下需要同步給module的消息
					CommonSettings commonSettings = CommonSettings.loadAllSettingsFromDB();
					SyncModuleInfosMessage infos = new SyncModuleInfosMessage();
					infos.setCommonSettings(commonSettings);
					
					for (BasicServiceModule module : modules) {
						GroupMessage msg = new GroupMessage();
						msg.setDestModuleID(module.getModuleName());
						msg.setMessageType(MessageTypes.S_SERVER_SYNC_INFOS);
						msg.setMessageContent(infos);
						MessageAction.sendMessage(msg);
					}

					serverReadyNotified = true;
				}
			}
		}
	}

	/**
	 * 添加组员信息
	 * 
	 * @param address
	 * @param gm
	 * 
	 * @author lipeilin
	 * @2011-3-10 下午05:54:24
	 */
	private void addMemberInfo(Address address, GroupMember gm) {
		GroupMember old = members.get(address);
		
		members.put(address, gm);
		

		// 添加新的
		List<String> ma = gm.getServiceModule();
		for (String moduleID : ma) {
			if (!moduleAddress.containsKey(moduleID)) {
				if (log.isInfoEnabled()) {
					int currentCount = moduleAddress.size() + 1;

					StringBuilder sb = new StringBuilder("Group add new member: " + currentCount + "/"
							+ ServerProperties.getTotalModuleCount() + " " + moduleID + " address:" + address);

					// 顯示剩下還沒加入的 module name
					if (ServerProperties.getTotalModuleCount() - currentCount < 20) {
						sb.append(" remain:");
						for (String moduleName : ServerProperties.getModulenamemap()) {
							if (!moduleAddress.containsKey(moduleName)) {
								sb.append(" ").append(moduleName);
							}
						}
					}

					log.info(sb.toString());
				}

//				if (moduleID.startsWith(ModuleName.ACCOUNT)) {
//					List<BasicServiceModule> list = new ArrayList<BasicServiceModule>(ModuleServer.getInstance()
//							.getModules());
//					for (BasicServiceModule module : list) {
//						if (module.getModuleName().startsWith(ModuleName.GATEWAY)
//								&& module instanceof GatewayModule) {
//							GatewayModule gatewayModule = (GatewayModule) module;
//							gatewayModule.notifyAccount();
//						}
//					}
//				}
			}
			moduleAddress.put(moduleID, address);
			
			// 加入moduleClassMap
			// "_"以前是module class名稱
			String moduleClassName = moduleID.split("_")[0];
			Set<String> moduleNameSet = moduleClassMap.get(moduleClassName);
			if (moduleNameSet == null) {
				moduleNameSet = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>(100));
			}
			moduleNameSet.add(moduleID);
			moduleClassMap.put(moduleClassName, moduleNameSet);
		}

		// 把多余的删了！
		if (old != null) {
			List<String> list = old.getServiceModule();
			for (String moduleID : list) {
				if (!gm.getServiceModule().contains(moduleID)) {
					moduleAddress.remove(moduleID);
					log.info("Group remove member:" + moduleID);
					
					// 移除moduleClassMap中的moduleName
					// "_"以前是module class名稱
					String moduleClassName = moduleID.split("_")[0];
					Set<String> moduleIDSet = moduleClassMap.get(moduleClassName);
					if (moduleIDSet != null) {
						moduleIDSet.remove(moduleID);
						if (moduleIDSet.isEmpty() == true) {
							moduleClassMap.remove(moduleClassName);
						}
					}
				}

			}
		}
		
		notifyServerReady();
	}

	/**
	 * 更新本地成员信息，并广播给组成员
	 * 
	 * 
	 * @author lipeilin
	 * @2011-3-10 下午06:07:25
	 */
	public GroupMember updateLocalMemberInfo() {
		// 添加本地信息
		Address address = channel.getAddress();
		List<BasicServiceModule> Modules = new ArrayList<BasicServiceModule>(moduleServer.getModules());
		GroupMember gm = new GroupMember();
		gm.setAddress(address.toString());
		gm.setServerName(moduleServer.getName());
		gm.setLoginTime(System.currentTimeMillis());
		for (BasicServiceModule module : Modules) {
			gm.getServiceModule().add(module.getModuleName());
		}
		addMemberInfo(address, gm);

		//sendLocalMemberInfo();
		return gm;
	}
	
	/**
	 * 檢查指定的module是否存在當前group裡面
	 * @param moduleName
	 * 以ModuleName設定的名稱為主
	 * @see com.xpec.c4.service.module.ModuleName
	 * @return
	 */
	public boolean isExistModule(String moduleName){
		List<GroupMember> allAddr = getMembers();
		if (allAddr != null) {
			for (GroupMember it : allAddr) {
				List<String> moduleList = it.getServiceModule();
				if (moduleList != null) {
					for (String m : moduleList) {
						if (m.startsWith(moduleName)) {
							return true;
						}
					}
				}
			}
		}
		return false;
	}	

	
	
//	/**
//	 * 通知Module作對應的邏輯<br>
//	 * 不過理論上底層不應該牽涉到遊戲邏輯 
//	 */
//	public void onServerChanged(){
//		GroupMessage message = new GroupMessage();
//		message.setSrcService(moduleServer.getName());
//		message.setMessageType(MessageTypes.S_ON_SERVER_CHANGED);
//		message.setMessageContent(null);
//		sendMessage(message);		
//	}

	public void stopGroupTransport() {

		channel.close();
		dispatcher.stop();

		running = false;
	}

	/**
	 * 发送信息，不等待返回
	 * 
	 * @param obj
	 * 
	 * @author lipeilin
	 * @2011-3-10 下午02:59:31
	 */
	public void sendMessage(GroupMessage message)
			throws GroupTransportException {
		
		//log.error("TransportManager  SendMessage - destModuleID:"+message.getDestModuleID() + " srcService:"+message.getSrcService() );
		
		
		RequestOptions options = new RequestOptions();
		options.setMode(ResponseMode.GET_NONE);

		sendMessage(message, options);
	}
	
	/**
	 * 发送信息，并等待返回	 
	 * @param message
	 * @param timeout 最長等多久(milliseconds), null代表一直等下去
	 * @return
	 * @throws GroupTransportException
	 */
	 
	public Object sendMessageWaitResponse(GroupMessage message, Long timeout)
			throws GroupTransportException {

		RequestOptions options = new RequestOptions();
//		options.setMode(ResponseMode.GET_FIRST);
		options.setMode(ResponseMode.GET_NONE);

		SyncSignalManager.SyncSignalObject syncSignal = syncSignalManager.getSyncSignal();
		message.setSyncId(syncSignal.signalId);
		sendMessage(message, options);
		
		syncSignalManager.waitSyncSignal(syncSignal, timeout);
		if(!syncSignal.getHasResult()) {
			syncSignalManager.releaseSyncSignal(syncSignal);
			throw new com.xpec.c4.service.exceptions.TimeoutException();
		}
		Object returnObj = syncSignal.returnObject;
		syncSignalManager.releaseSyncSignal(syncSignal);
		
		return returnObj;
	}

	/**
	 * 
	 * 发送本地信息给其他成员
	 * 
	 * @author lipeilin
	 * @2011-4-23 下午05:41:37
	 */
	public void sendLocalMemberInfo() {
		// 把本地信息发送出去
		GroupMessage message = new GroupMessage();
		message.setSrcService(moduleServer.getName());
		message.setMessageType(MessageTypes.SEND_SERVER_INFO);
		message.setMessageContent(getLocal());
		sendMessage(message);
	}

	/**
	 * 发送消息
	 * 
	 * @param message
	 * @param options
	 * @return
	 * 
	 * @author lipeilin
	 * @2011-3-10 下午04:06:38
	 * @throws Exception
	 */
	public Object sendMessage(GroupMessage message, RequestOptions options) {
		if (!running)
			return null;

		//節點內傳送，檢查srcservice 是否為空，為空值才做getname的設定
		if(message.getSrcService()=="")
		{
			message.setSrcService(moduleServer.getName());
		}

		// 返回的结果
		Object response = null;

		// 需要传送的内容
		String destModuleID = message.getDestModuleID();
		boolean toClient = message.isToClient();
		int messageType = message.getMessageType();
		int syncId = message.getSyncId();
		String srcService = message.getSrcService();

		Collection<GUID> sessionIDs = message.getSessionIDs();
		if (sessionIDs == null) {
			sessionIDs = new ArrayList<GUID> ();
		}

		GUID sessionID = message.getSessionID();
		if (sessionID != null) {
			sessionIDs.add(sessionID);
		}

		// 不填写地址，就是广播
		if (destModuleID == null || destModuleID.length() == 0) {
			// 是有指定某些 module 才會送的 message
			if (broadcastMsgListenModuleMap.containsKey(messageType)) {
				HashSet<String> moduleClassSet = broadcastMsgListenModuleMap.get(messageType);

				for (String moduleClassName : moduleClassSet) {
					Set<String> moduleIDSet = moduleClassMap.get(moduleClassName);
					if (moduleIDSet != null) {
						for (String moduleID : moduleIDSet) {
							Address addr = moduleAddress.get(moduleID);
							if (addr != null) {
								// 改destModuleID為目標moduleID 才能正確送達該module
								message.setDestModuleID(moduleID);
								Message msg = new Message(addr, null, message.encode());
								try {
									dispatcher.sendMessageWithFuture(msg, options);
								} catch (Exception e) {
									log.error("send message error!", e);
								}
							}
						}
						// 將destModuleID設回原來moduleClass 避免意外情形
						message.setDestModuleID(destModuleID);
					} // if (moduleIDSet != null) {
				} // for (String moduleClassName : moduleClassSet) {
			} else {
				// 一般的廣播消息
				List<Address> allAddress = getAllAddress();

				for (Address address : allAddress) {
					Message msg = new Message(address, null, message.encode());
					try {
						dispatcher.sendMessageWithFuture(msg, options);
					} catch (Exception e) {
						log.error("send message error!", e);
					}
				}
			}
		} else // 填写地址，则发送给指定地址
		{
			Address address = moduleAddress.get(destModuleID);

			if (address != null) {
				// 如果是本地，则直接函数调用
				if (address.equals(channel.getAddress())) {
					// 发送给client的
					if (toClient) {
						//廣播訊息要特殊處理，因為原本廣播是在handle處理的						
						if(sessionIDs == null || sessionIDs.size() == 0)
						{
							if(message.isBroadCast())
							{
								//Rowan: may be not thread safe??
								BroadcastTask.execute(new BroadcastTask(message));
							}
						}
						else
						{												
							byte[] messageContent = message.getMessageContentBytes();
							//rowan: intercept bytes to clients here
							MessageAction.sendClientMessageBytes(sessionIDs,
									messageType, messageContent);
						}
					}
					// 如果是异步,则通过Timer调用
					else if (options.getMode() == ResponseMode.GET_NONE) {
						try {
							response = MessageHandle.handleMessage(messageType, syncId,
									message.getMessageContentBytes(), sessionID,
									destModuleID, srcService, channel.getAddress());
						} catch (Exception e) {
							log.error("handle message error!", e);
						}
	
					}
					// 如果是同步,则直接调用函数
					else {
						try {
							response = MessageHandle.handleMessage(messageType, syncId,
									message.getMessageContentBytes(), sessionID,
									destModuleID, srcService, channel.getAddress());
						} catch (Exception e) {
							log.error("handle message error!", e);
						}
					}
				} else // 如果是集群内其他节点服务器，节点发送
				{
					Message msg = new Message(address, null, message.encode());
					try {
						// 如果不需要返回值,用异步
						if (options.getMode() == ResponseMode.GET_NONE) {
							dispatcher.sendMessageWithFuture(msg, options);
						}
						// 否则用同步
						else {
							Object obj = dispatcher.sendMessage(msg, options);
							if (obj != null) {
								response = CodecUtility.decodeByte((byte[]) obj);
							}
						}
					} catch (TimeoutException e) {
						log.error("sendMessage error!" + message, e);
					} catch (SuspectedException e) {
						log.error("sendMessage error!" + message, e);
					} catch (Exception e) {
						log.error("decode message byte error!", e);
					}
				}
			}
			else {
				// 填的是moduleClass 則同種類的module廣播
				Set<String> moduleIDSet = moduleClassMap.get(destModuleID);
				if (moduleIDSet != null) {
					for (String moduleID : moduleIDSet) {
						Address addr = moduleAddress.get(moduleID);
						if (addr != null) {
							// 改destModuleID為目標moduleID 才能正確送達該module
							message.setDestModuleID(moduleID);
							Message msg = new Message(addr, null, message.encode());
							try {
								dispatcher.sendMessageWithFuture(msg, options);
							} catch (Exception e) {
								log.error("send message error!", e);
							}
						}
					}
					// 將destModuleID設回原來moduleClass 避免意外情形
					message.setDestModuleID(destModuleID);
				}
				else {
					// moduleAddress及moduleClassMap都找不到
					String error = destModuleID
							+ " address and moduleClass is not exist!!!"
							+ " messageType:" + messageType
							+ " sessionID:" + sessionID;
					log.error(error, new GroupTransportException(error));
					return null;
				}
				
			}
		}
		return response;
	}

	/**
	 * 发送消息
	 * 
	 * @param address
	 * @param message
	 * @param options
	 * 
	 * @author lipeilin
	 * @2011-3-15 下午12:06:35
	 */
	public void sendMessage(Address address, GroupMessage message,
			RequestOptions options) {

		byte[] content = message.encode();

		Message msg = new Message(address, null, content);

		try {
			dispatcher.sendMessageWithFuture(msg, options);
		} catch (Exception e) {
			log.error("sendMessage error!" + message, e);
		}
	}
	
	/**
	 * 送出回應
	 * @param address
	 * @param syncId
	 * @param returnObj
	 */
	public void sendResponse(Address address, int syncId, Object returnObj) {
		if (address == null || address == channel.getAddress()) {
			//本地端
			//激發等待中的執行緒並返回值
			syncSignalManager.notifySyncSignal(syncId, returnObj);
		}
		else {
			RequestOptions options = new RequestOptions();
			options.setMode(ResponseMode.GET_NONE);
			
			GroupMessage message = new GroupMessage();
			message.setSrcService(moduleServer.getName());
			message.setMessageType(MessageTypes.SYNC_RETURN);
			message.setSyncId(syncId);
			message.setMessageContent(returnObj);
			sendMessage(address, message, options);
		}
	}

	// public void putModuleIdName(Integer id, String name) {
	// moduleIdNames.put(id, name);
	// }
	//
	// public Address getModuleAddress(int moduleID) {
	// String moduleName = moduleIdNames.get(moduleID);
	// Address address = moduleAddress.get(moduleName);
	// return address;
	// }

	// public String getModuleName(int moduleID) {
	// String moduleName = moduleIdNames.get(moduleID);
	// return moduleName;
	// }
	//
	// public HashMap<Integer, String> getModuleIdName() {
	// return moduleIdNames;
	// }

	// public void setModuleIdName(HashMap<Integer, String> moduleIdName) {
	// this.moduleIdNames = moduleIdName;
	// }

	public static void main(String[] args) throws Exception {
		TransportManager manager = new TransportManager(
				ModuleServer.getInstance());
		manager.startGroupTransport();
	}

	class GroupMembershipListener implements MembershipListener {

		/*
		 * (non-Javadoc)
		 * 
		 * @see org.jgroups.MembershipListener#block()
		 */
		@Override
		public void block() {
			log.info("block");
		}

		/*
		 * (non-Javadoc)
		 * 
		 * @see org.jgroups.MembershipListener#suspect(org.jgroups.Address)
		 */
		@Override
		public void suspect(Address suspected_mbr) {
			log.error("suspect:" + suspected_mbr);
			// members.remove(suspected_mbr);
			// moduleAddress.remove(suspected_mbr.toString());

			// log.info("all members:"
			// + new HashMap<Address, GroupMember>(members));
			// log.info("all address:"
			// + new HashMap<String, Address>(moduleAddress));
		}

		/*
		 * (non-Javadoc)
		 * 
		 * @see org.jgroups.MembershipListener#viewAccepted(org.jgroups.View)
		 */
		@Override
		public void viewAccepted(View new_view) {
			// 不存在的删除
			List<Address> existAddress = new_view.getMembers();
			ArrayList<Address> oldAddress = new ArrayList<Address>(members.keySet());
			
			boolean onServerChanged = false;
			
			for (Address address : oldAddress) {
				// 如果不存在则，删除
				String addressString = address.toString();
				if (!existAddress.contains(address)) {
					GroupMember gm = members.get(address);
					members.remove(address);

					for (String moduleID : gm.getServiceModule()) {
						moduleAddress.remove(moduleID);
						
						if (log.isInfoEnabled()) {
							log.info("Group remove member: " + moduleAddress.size() + "/"
									+ ServerProperties.getTotalModuleCount() + " " + moduleID + " address:"
									+ address);
						}

						// 移除moduleClassMap中的moduleName
						// "_"以前是module class名稱
						String moduleClassName = moduleID.split("_")[0];
						Set<String> moduleIDSet = moduleClassMap.get(moduleClassName);
						if (moduleIDSet != null) {
							moduleIDSet.remove(moduleID);
							if (moduleIDSet.isEmpty() == true) {
								moduleClassMap.remove(moduleClassName);
							}
						}

//						// 
//						if (moduleID.startsWith(ModuleName.GATEWAY)) {
//							AccountCollectionInfo.getInstance()
//									.removeGatewayLoadingInfo(moduleID);
//						}
					}

					if (log.isInfoEnabled()) {
						log.info("Group remove address: " + addressString);
					}
					onServerChanged = true;
				}
			}
			//檢查是否有新加入的member , 把其address先加入(但沒有member, member要等它開好才會加進來)
			for(Address address : existAddress){
				//新加入的member
				if(!oldAddress.contains(address)){
					GroupMember gm = new GroupMember();
					gm.setAddress(address.toString());
					
					members.put(address, gm);
				}
			}
			if (log.isInfoEnabled()) {
				log.info("viewAccepted:" + new_view);
			}
			
			//if server is ready,then report
			if(ModuleServer.getInstance().isReady()){
				
				sendLocalMemberInfo();
				//TODO 不知道還要不要Task
				GroupMessage message = new GroupMessage();
				message.setSrcService(moduleServer.getName());
				message.setMessageType(MessageTypes.SEND_SERVER_INFO);
				message.setMessageContent(getLocal());

				MemberInfoSender task = new MemberInfoSender(existAddress, message);

				JobExecutor.schedule(task,
						10 * TimeConstants.MILLI_SECOND_PRE_SECOND);
				
				if (log.isInfoEnabled()) {
					log.info("send MemberInfoSender after 10 sec");
				}
			}
			
			notifyServerReady();
			
//			// 如有发现其它服务器进程关闭
//			if(onServerChanged){
//				onServerChanged();
//			}
		}

		/*
		 * (non-Javadoc)
		 * 
		 * @see org.jgroups.MembershipListener#unblock()
		 */
		@Override
		public void unblock() {
			log.info("unblock");
		}

	}

	class GroupMessageHandler implements RequestHandler {
		
		private ModuleServer moduleServer = null; 
		
		public GroupMessageHandler(ModuleServer moduleServer) {
			this.moduleServer = moduleServer;	
		}

		/*
		 * (non-Javadoc)
		 * 
		 * @see org.jgroups.blocks.RequestHandler#handle(org.jgroups.Message)
		 */
		@Override
		public Object handle(Message msg) 
		{
			byte[] bytes = msg.getBuffer();

			GroupMessage content = null;
			Object result = null;

			try 
			{
				content = new GroupMessage();
				content.decode(bytes);
				// 组成员管理信息
				if (content.getMessageType() == MessageTypes.SEND_SERVER_INFO)
				{
					onReceiveServerInfo(msg, content);
				}
				else if (content.getMessageType() == MessageTypes.SESSION_CONTEXT_CHANGE) 
				{
					ContextStatus message = (ContextStatus) CodecUtility.decodeByte(content.getMessageContentBytes());
					ContextsSessionTask task = new ContextsSessionTask(message);
					task.run();
					//會有順序風險, 先就地執行
					//JobExecutor.execute(task);
				}
				// user context改变信息
				else if (content.getMessageType() == MessageTypes.USERID_SESSIONID_SYNCHRO) 
				{
					ContextStatus message = (ContextStatus) CodecUtility.decodeByte(content.getMessageContentBytes());
					Contexts.synchroUserSession(message.getUserId(),message.getSessionId());
				}
				else if (content.getMessageType() == MessageTypes.SERVER_ABOUT_TO_CLOSE)
				{
					if(content.getSrcService().equals(moduleServer.getName()))
					{
						onReceiveServerToClose(content);
					}
				}
				// session强制关闭
				else if (content.getMessageType() == MessageTypes.SESSION_CLOSE) 
				{
					SessionCloseNotify nofity = (SessionCloseNotify) CodecUtility.decodeByte(content.getMessageContentBytes());
					GUID sessionID = nofity.getSessionID();
					
					if (sessionID == null) {
						if (log.isInfoEnabled()) {
							log.info("Session Close Log " +  GameUtility.getCurMethodName() + " SESSION_CLOSE START! SessionId = null");
						}
					}
					
					CloseType closeType = nofity.getCloseType();
					
					//斷線處理流程
					if (ServerProperties.isTargetModule(PlayerCommonInfoManager.getInstance().getGatewayBySessionId(sessionID), ModuleName.H5GATEWAY)) {
						H5ServerOperations.disconnectionAllGateWayProcess(sessionID);
					} else {
						ServerOperations.disconnectionAllGateWayProcess(sessionID);
					}
					
					if (sessionID == null) {
						if (log.isInfoEnabled()) {
							log.info("Session Close Log " +  GameUtility.getCurMethodName() + " SESSION_CLOSE End! SessionId = null");
						}
					}
					
					XSENetSession session = ServiceInfo.getSession(sessionID);
					// 有的话，就是本地关闭就行。
					if (session != null) {
						session.getParameters().put(
								ServerOperations.CLOSE_TYPE, closeType);
						session.close();
						log.info("kick session:" + sessionID);
					} else if (ServerProperties.isTargetModule(PlayerCommonInfoManager.getInstance().getGatewayBySessionId(sessionID), ModuleName.H5GATEWAY)) {
						H5XSENetSession h5Session = H5ServiceInfo.getSession(sessionID);
						if (h5Session != null) {
							h5Session.getParameters().put(H5ServerOperations.CLOSE_TYPE, closeType);
							h5Session.close();
							log.info("kick h5session:" + sessionID);
						}
					}
				}
				else if (content.getMessageType() == MessageTypes.SYNC_RETURN) 
				{
					//激發等待中的執行緒並返回值
					Object returnObj = null;
					try 
					{
						returnObj = CodecUtility.decodeByte(content.getMessageContentBytes());
					} 
					catch (Throwable t) 
					{
						throw t;
					} 
					finally 
					{
						int syncId = content.getSyncId();
						syncSignalManager.notifySyncSignal(syncId, returnObj);
					}
				}
//				else if(content.getMessageType() == MessageTypes.TELNET_MANAGER){
//					Object returnObj = CodecUtility.decodeByte(content
//							.getMessageContentBytes());
//					if (returnObj instanceof STelnetCmdResult) {
//						TelnetConnectManager.getInstance()
//								.processCommandResult(
//										(STelnetCmdResult) returnObj);
//					}
//				}
				// 是否由网关发给client
				else if (content.isToClient()) 
				{
					// 群发
					Collection<GUID> ids = content.getSessionIDs();
					if (ids != null && ids.size() > 0) 
					{
						
						MessageAction.sendClientMessageBytes(ids,
								content.getMessageType(),
								content.getMessageContentBytes());
					}
					else if(content.isBroadCast())
					{
						//Rowan: may be not thread safe??
						BroadcastTask.execute(new BroadcastTask(content));
					}
					else if(content.getChannelID() > 0)
					{
						//頻道內群發
					}
							
				} 
				else 
				{
					GUID sessionID = content.getSessionID();
					if (sessionID == null && content.getSessionIDs() != null) 
					{
						sessionID = ((List<GUID>) content.getSessionIDs()).get(0);
					}

					// 如果sessionID不是null，表示由client发送过来的，不用等返回值，交给线程池处理
					if (sessionID != null) 
					{
						MessageHandle.handleMessage(content.getMessageType(), content.getSyncId(),
								content.getMessageContentBytes(), sessionID,
								content.getDestModuleID(),
								content.getSrcService(),
								msg.getSrc());

					}
					// 否则是模块之间的发送
					else {
						result = MessageHandle.handleMessage(
								content.getMessageType(),
								content.getSyncId(),
								content.getMessageContentBytes(), sessionID,
								content.getDestModuleID(),
								content.getSrcService(),
								msg.getSrc());
					}

					if (result != null) {
						result = CodecUtility.encodeByte(result,
								content.getMessageType());
					}
				}
			} 
			catch (Exception e) 
			{
				log.error("cannot handle message:" + content, e);
			} 
			catch (Throwable fatal) 
			{
				String err = String.format("fatal error : cannot handle message: %s", content);
				log.fatal(err, fatal);
			}

			return result;
		}

		/**
		 * ASYNC close all module and etc...
		 * @param content
		 */
		private void onReceiveServerToClose(GroupMessage content) 
		{			
			moduleServer.stop();
			moduleServer.createSignalFile(false);
			//self quit
			System.exit(0);
		}

		private boolean onReceiveServerInfo(Message msg, GroupMessage content) {
			{	
				GroupMember member = (GroupMember) CodecUtility
						.decodeByte(content.getMessageContentBytes());

				initLocalMemberInfo();

				// 如果server或module已经存在，则不能添加，并错误处理
				if (!msg.getSrc().equals(channel.getAddress())) {
					Collection<GroupMember> memberList = new ArrayList<GroupMember>(
							members.values());
					for (GroupMember groupMember : memberList) {

						// 如果是同一server的信息，则跳过
						if (groupMember.getAddress().equals(
								member.getAddress())) {
							break;
							// return null;
						}
						//有可能是view accept 先加上去的, 那就不要檢查了
						if(groupMember.getServerName() == null)
							continue;
						if (groupMember.getServerName().equals(
								member.getServerName())) {
							log.error("*********** "
									+ member.getServerName()
									+ " server already exist!"
									+ " ***********");
							// 如果别人先加入的，则断开自己
							if (member.getLoginTime() < getLocal()
									.getLoginTime()) {
								log.error("********** "
										+ "shut down now! **********");
								System.exit(0);
							}
							return false;
						}
						for (String module : groupMember.getServiceModule()) {
							if (member.getServiceModule().contains(module)) {
								log.error("*********** " + module
										+ " module already exist!"
										+ " ***********");
								// 如果别人先加入的，则断开自己
								if (member.getLoginTime() < getLocal()
										.getLoginTime()) {
									log.error("********** "
											+ "shut down now! **********");
									System.exit(0);
								}
								return false;
							}
						}
					}
				}
				else{
					//自己送給自己? 浪費效能
					return false;
				}
				// 添加组成员信息
				addMemberInfo(msg.getSrc(), member);
				return true;
			}
		}
	}

	public boolean isRunning() {
		return running;
	}

	public void setRunning(boolean running) {
		this.running = running;
	}

	/**
	 * 获得全部连接的模块列表
	 * 
	 * @return
	 * 
	 * @author lipeilin
	 * @2011-3-24 下午05:16:07
	 */
	public List<String> getModuleIdList() {
		return new ArrayList<String>(moduleAddress.keySet());
	}

	/**
	 * 是否存在这个模块？
	 * 
	 * @param moduleID
	 * @return true，存在;false，不存在
	 * 
	 * @2011-11-11 下午03:34:20
	 */
	public boolean hasModule(String moduleID) {
		if(moduleID == null)
			return false;
		return moduleAddress.containsKey(moduleID);
	}
	
	/**
	 * 是否存在此類別的模塊
	 * @param moduleClassID
	 * @return
	 */
	public boolean hasModuleClass(String moduleClassID)
	{
		return moduleClassMap.containsKey(moduleClassID);
	}

	public void sendShutDownMessage() {
		GroupMessage message = new GroupMessage();
		message.setSrcService(moduleServer.getName());
		message.setMessageType(MessageTypes.SERVER_ABOUT_TO_CLOSE);
		message.setMessageContent("close");
		sendMessage(message);		
	}

	/**
	 * 取得所有的 server address 底下的其中一個 module<br>
	 * 可以用這個列表理的 module id 將消息送給所有的 vm
	 * 
	 * @return
	 */
	public Collection<String> getOneModuleInAllAddress() {
		HashMap<Address, String> tempAddressMap = new HashMap<Address, String>();

		for (Map.Entry<String, Address> entry : moduleAddress.entrySet()) {
			String moduleId = entry.getKey();
			Address address = entry.getValue();
			tempAddressMap.put(address, moduleId);
		}

		return tempAddressMap.values();
	}
	
	/**
	 * 取得所有本地 (目前vm) 的 module 狀態
	 */
	public ArrayList<ModuleStatus> getLocalModuleStatusList() {
		ArrayList<ModuleStatus> moduleStatusList = new ArrayList<ModuleStatus>();

		Address localAddress = channel.getAddress();

		// 掃一遍所有的 module id
		for (Map.Entry<String, Address> entry : moduleAddress.entrySet()) {
			String moduleId = entry.getKey();
			Address address = entry.getValue();

			// 是本地的 module
			if (address.equals(localAddress)) {
				ModuleStatus moduleStatus = new ModuleStatus();

				moduleStatus.setModuleId(moduleId);

				BasicServiceModule module = moduleServer.getBasicServiceModule(moduleId);
				if (module != null) {
					// module 存在
					moduleStatus.setActive(true);
				} else {
					// module 不存在
					moduleStatus.setActive(false);
				}

				moduleStatusList.add(moduleStatus);
			}
		}

		return moduleStatusList;
	}
	
	/**
	 * 檢查指定的 module id 是否為本地 (同一個vm) module
	 * 
	 * @param moduleId
	 * @return
	 */
	public boolean isLocalModule(String moduleId) {
		if (!GameUtility.isHaveString(moduleId)) {
			return false;
		}

		// 本地 address
		Address localAddr = channel.getAddress();

		// module 所在 address
		Address addr = moduleAddress.get(moduleId);

		// 相同就本地
		if (addr.equals(localAddr)) {
			return true;
		}

		return false;
	}
}
